/*
Copyright The Helm Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

Original file: https://github.com/helm/helm/blob/v3.2.0/pkg/kube/wait.go
*/

package kube

import (
	"context"
	"fmt"
	"strings"
	"time"

	"github.com/pkg/errors"
	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	appsv1 "k8s.io/api/apps/v1"
	appsv1beta1 "k8s.io/api/apps/v1beta1"
	appsv1beta2 "k8s.io/api/apps/v1beta2"
	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	extensionsv1beta1 "k8s.io/api/extensions/v1beta1"
	apiextv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
	apiextv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/util/intstr"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/kubernetes"

	bkev1beta1 "gopkg.openfuyao.cn/cluster-api-provider-bke/api/v1beta1"
	"gopkg.openfuyao.cn/cluster-api-provider-bke/utils"
	"gopkg.openfuyao.cn/cluster-api-provider-bke/utils/deploymentutil"
	"gopkg.openfuyao.cn/cluster-api-provider-bke/utils/log"
)

const (
	WaitTimeout  = 4 * time.Minute
	WaitInterval = 2 * time.Second
)

type waiter struct {
	// The object to wait for.
	unstructuredObj *unstructured.Unstructured
	namespace       string
	name            string
	block           bool
	timeout         time.Duration
	interval        time.Duration
	checker         *readyChecker
	ctx             context.Context
}

type readyChecker struct {
	client        *kubernetes.Clientset
	dynamicClient dynamic.Interface
	log           *zap.SugaredLogger
	bkeLog        *bkev1beta1.BKELogger
	pausedAsReady bool
	fullComplete  bool
}

func (c *Client) Wait(obj *unstructured.Unstructured, task *Task) error {
	return newWaiterFromUnstructured(obj).
		setPoller(task).
		setChecker(c).
		wait()
}
func (w *waiter) setPoller(task *Task) *waiter {
	w.timeout = task.Timeout
	w.interval = task.Interval
	w.block = task.Block
	return w
}

func (w *waiter) setChecker(c *Client) *waiter {
	w.checker = newCheckerFromKubeClient(c)
	w.checker.fullComplete = w.block
	w.ctx = c.Ctx
	return w
}

func (w *waiter) wait() error {
	ctx, cancel := context.WithTimeout(w.ctx, w.timeout)
	defer cancel()
	return wait.PollImmediateUntil(w.interval, func() (bool, error) {
		ready, err := w.checker.isReady(ctx, w)
		if !ready || err != nil {
			if err != nil {
				if isServiceUnavailable(err) {
					return false, nil
				}
				log.Warnf("Waiting for %s/%s, err: %v", w.namespace, w.name, err)
			}
			return false, nil
		}
		return true, nil
	}, ctx.Done())
}

func newCheckerFromKubeClient(c *Client) *readyChecker {
	return &readyChecker{
		client:        c.ClientSet,
		dynamicClient: c.DynamicClient,
		log:           c.Log,
		bkeLog:        c.BKELog,
		pausedAsReady: true,
	}
}
func newWaiterFromUnstructured(obj *unstructured.Unstructured) *waiter {
	return &waiter{
		unstructuredObj: obj,
		namespace:       obj.GetNamespace(),
		name:            obj.GetName(),
	}
}

func (c *readyChecker) isReady(ctx context.Context, w *waiter) (bool, error) {
	switch w.unstructuredObj.GetObjectKind().GroupVersionKind().Kind {
	case "Deployment":
		currentDeployment, err := c.client.AppsV1().Deployments(w.namespace).Get(ctx, w.name, metav1.GetOptions{})
		if err != nil {
			return false, err
		}
		// If paused deployment will never be ready
		if currentDeployment.Spec.Paused {
			return c.pausedAsReady, nil
		}
		// Find RS associated with deployment
		newRepSet, err := deploymentutil.GetNewReplicaSet(currentDeployment, c.client.AppsV1())
		if err != nil || newRepSet == nil {
			return false, err
		}
		if !c.deploymentReady(newRepSet, currentDeployment) {
			return false, nil
		}
	case "PersistentVolumeClaim":
		clm, err := c.client.CoreV1().PersistentVolumeClaims(w.namespace).Get(ctx, w.name, metav1.GetOptions{})
		if err != nil {
			return false, err
		}
		if !c.volumeReady(clm) {
			return false, nil
		}
	case "Service":
		service, err := c.client.CoreV1().Services(w.namespace).Get(ctx, w.name, metav1.GetOptions{})
		if err != nil {
			return false, err
		}
		if !c.serviceReady(service) {
			return false, nil
		}
	case "DaemonSet":
		daemonSet, err := c.client.AppsV1().DaemonSets(w.namespace).Get(ctx, w.name, metav1.GetOptions{})
		if err != nil {
			return false, err
		}
		if !c.daemonSetReady(daemonSet) {
			return false, nil
		}
	case "StatefulSet":
		state, err := c.client.AppsV1().StatefulSets(w.namespace).Get(ctx, w.name, metav1.GetOptions{})
		if err != nil {
			return false, err
		}
		if !c.statefulSetReady(state) {
			return false, nil
		}
	case "ReplicaSet":
		var ok, err = c.podsReadyForObject(ctx, w.namespace, w.unstructuredObj)
		if !ok || err != nil {
			return false, err
		}
	case "Paas":
		var ok, err = c.paasReady(ctx, w.unstructuredObj)
		if !ok || err != nil {
			return false, err
		}
		return true, nil
	}
	return true, nil
}

// ProductStatus defines the desired state of product status
type ProductStatus struct {
	Name           string            `json:"name"`
	StartTime      *metav1.Time      `json:"startTime,omitempty"`
	UpdateTime     *metav1.Time      `json:"updateTime,omitempty"`
	CompletionTime *metav1.Time      `json:"completionTime,omitempty"`
	Health         bool              `json:"health"`
	Component      []ComponentStatus `json:"component,omitempty"`
	Reason         string            `json:"reason"`
}

// ComponentStatus defines the desired state of component status
type ComponentStatus struct {
	Name     string `json:"name"`
	Resource string `json:"resource"`
	Health   bool   `json:"componentHealth"`
	Message  string `json:"message"`
}

var paasProductComponentStatus map[string]map[string]bool

func (c *readyChecker) paasReady(ctx context.Context, unstructuredObj *unstructured.Unstructured) (paasReady bool, err error) {
	// build gvr
	group := unstructuredObj.GroupVersionKind().Group
	version := unstructuredObj.GetObjectKind().GroupVersionKind().Version
	kind := unstructuredObj.GetObjectKind().GroupVersionKind().Kind
	namespace := unstructuredObj.GetNamespace()
	name := unstructuredObj.GetName()

	ressource := ""
	if strings.HasSuffix(kind, "s") {
		ressource = strings.ToLower(kind)
	} else {
		ressource = strings.ToLower(kind + "s")
	}
	gvr := schema.GroupVersionResource{
		Group:    group,
		Version:  version,
		Resource: ressource,
	}
	paasObj, err := c.dynamicClient.Resource(gvr).Namespace(namespace).Get(ctx, name, metav1.GetOptions{})
	if err != nil {
		return false, err
	}

	ready, found, err := unstructured.NestedFieldCopy(paasObj.Object, "status", "health")
	if err != nil || !found {
		return false, err
	}
	c.log.Infof("paas %q state: %v", utils.ClientObjNS(paasObj), ready)

	defer func() {
		if err := recover(); err != nil {
			c.log.Warn("get paas %q status panic: %v", utils.ClientObjNS(paasObj), err)
			paasReady = ready.(bool)
			return
		}
	}()

	if c.bkeLog != nil {
		// log paas component status
		productStatus, found, err := unstructured.NestedFieldCopy(paasObj.Object, "status", "products")
		if err != nil || !found {
			return false, err
		}

		if paasProductComponentStatus == nil {
			paasProductComponentStatus = make(map[string]map[string]bool)
		}
		// convert product status to []ProductStatus
		productStatusList := productStatus.([]interface{})
		for _, productInterface := range productStatusList {
			productTmp := productInterface.(map[string]interface{})

			productName := productTmp["name"].(string)

			if paasProductComponentStatus[productName] == nil {
				paasProductComponentStatus[productName] = make(map[string]bool)
			}
			productComponentList := productTmp["component"].([]interface{})
			for _, componentInterface := range productComponentList {
				componentTmp := componentInterface.(map[string]interface{})

				componentName := componentTmp["name"].(string)
				componentHealth := componentTmp["componentHealth"].(bool)
				componentMessage := componentTmp["message"].(string)

				//第一次记录，打印状态
				if _, ok := paasProductComponentStatus[productName][componentName]; !ok {
					paasProductComponentStatus[productName][componentName] = componentHealth
					c.bkeLog.Info("PortalDeploying", "paas product %q sub component %q state: %v with message: %v", productName, componentName, componentHealth, componentMessage)
					continue
				}

				// 如果本次状态和上次状态一致，忽略
				if paasProductComponentStatus[productName][componentName] == componentHealth {
					continue
				}

				// 如果本次状态和上次状态不一致，打印状态
				paasProductComponentStatus[productName][componentName] = componentHealth
				c.bkeLog.Info("PortalDeploying", "paas product %q sub component %q state: %v with message: %v", productName, componentName, componentHealth, componentMessage)
			}
		}
	}

	// ready is bool
	return ready.(bool), nil
}

func (c *readyChecker) podsReadyForObject(ctx context.Context, namespace string, obj runtime.Object) (bool, error) {
	podsForOb, err := c.podsforObject(ctx, namespace, obj)
	if err != nil {
		return false, err
	}
	for _, pod := range podsForOb {
		if !c.isPodReady(&pod) {
			return false, nil
		}
	}
	return true, nil
}

func (c *readyChecker) podsforObject(ctx context.Context, namespace string, obj runtime.Object) ([]corev1.Pod, error) {
	selectors, err := SelectorsForObject(obj)
	if err != nil {
		return nil, err
	}
	list, err := getPods(ctx, c.client, namespace, selectors.String())
	return list, err
}

func (c *readyChecker) isPodReady(p *corev1.Pod) bool {
	for _, con := range p.Status.Conditions {
		if con.Type == corev1.PodReady && con.Status == corev1.ConditionTrue {
			return true
		}
	}
	c.log.Infof("Pod is not ready: %s/%s", p.GetNamespace(), p.GetName())
	return false
}

func (c *readyChecker) deploymentReady(repSet *appsv1.ReplicaSet, dep *appsv1.Deployment) bool {
	if c.fullComplete {

		return *dep.Spec.Replicas == dep.Status.AvailableReplicas
	}
	expectedReady := *dep.Spec.Replicas - deploymentutil.MaxUnavailable(*dep)
	if !(repSet.Status.ReadyReplicas >= expectedReady) {
		c.log.Infof("Deployment is not ready: %s/%s. %d out of %d expected pods are ready",
			dep.Namespace, dep.Name, repSet.Status.ReadyReplicas, expectedReady)
		return false
	}
	return true
}

func isServiceUnavailable(err error) bool {
	if err != nil {
		err = rpctypes.Error(err)
		if e, ok := err.(rpctypes.EtcdError); ok {
			if e.Code() == codes.Unavailable {
				return true
			}
		}
	}
	return false
}

func (c *readyChecker) volumeReady(volume *corev1.PersistentVolumeClaim) bool {
	if volume.Status.Phase != corev1.ClaimBound {
		c.log.Infof("PersistentVolumeClaim is not bound: %s/%s", volume.GetNamespace(), volume.GetName())
		return false
	}
	return true
}

func (c *readyChecker) serviceReady(svc *corev1.Service) bool {
	// ExternalName Services are external to cluster so helm shouldn't be checking to see if they're 'ready' (i.e. have an IP Set)
	if svc.Spec.Type == corev1.ServiceTypeExternalName {
		return true
	}

	// Ensure that the service cluster IP is not empty
	if svc.Spec.ClusterIP == "" {
		c.log.Infof("Service does not have cluster IP address: %s/%s", svc.GetNamespace(), svc.GetName())
		return false
	}

	// This checks if the service has a LoadBalancer and that balancer has an Ingress defined
	if svc.Spec.Type == corev1.ServiceTypeLoadBalancer {
		// do not wait when at least 1 external IP is set
		if len(svc.Spec.ExternalIPs) > 0 {
			c.log.Infof("Service %s/%s has external IP addresses (%v), marking as ready",
				svc.GetNamespace(), svc.GetName(), svc.Spec.ExternalIPs)
			return true
		}

		if svc.Status.LoadBalancer.Ingress == nil {
			c.log.Infof("Service does not have load balancer ingress IP address: %s/%s", svc.GetNamespace(), svc.GetName())
			return false
		}
	}

	return true
}

func (c *readyChecker) daemonSetReady(daeSet *appsv1.DaemonSet) bool {
	if c.fullComplete {
		if daeSet.Status.DesiredNumberScheduled == 0 {
			return false
		}
		c.log.Infof("DaemonSet is not available: %s/%s. %d out of %d expected pods are available",
			daeSet.Namespace, daeSet.Name, daeSet.Status.NumberAvailable, daeSet.Status.DesiredNumberScheduled)
		return int(daeSet.Status.DesiredNumberScheduled) == int(daeSet.Status.NumberAvailable)
	}
	// If the update strategy is not a rolling update, there will be nothing to wait for
	if daeSet.Spec.UpdateStrategy.Type != appsv1.RollingUpdateDaemonSetStrategyType {
		return true
	}
	// Make sure all the updated pods have been scheduled
	if daeSet.Status.UpdatedNumberScheduled != daeSet.Status.DesiredNumberScheduled {
		c.log.Infof("DaemonSet is not ready: %s/%s. %d out of %d expected pods have been scheduled",
			daeSet.Namespace, daeSet.Name, daeSet.Status.UpdatedNumberScheduled, daeSet.Status.DesiredNumberScheduled)
		return false
	}
	// Make sure the status is up-to-date with the DaemonSet changes
	var maxUnavailable, err = intstr.GetValueFromIntOrPercent(
		daeSet.Spec.UpdateStrategy.RollingUpdate.MaxUnavailable, int(daeSet.Status.DesiredNumberScheduled), true)
	if err != nil {
		// If the value is invalid for any reason, set maxUnavailable to the number of desired replicas.
		// This matches the behavior of the `MaxUnavailable` function in deploymentutil.
		maxUnavailable = int(daeSet.Status.DesiredNumberScheduled)
	}
	// Calculate the expected number of ready pods for the DaemonSet
	var expectedReady = int(daeSet.Status.DesiredNumberScheduled) - maxUnavailable
	if !(int(daeSet.Status.NumberReady) >= expectedReady) {
		c.log.Infof("DaemonSet is not ready: %s/%s. %d out of %d expected pods are ready",
			daeSet.Namespace, daeSet.Name, daeSet.Status.NumberReady, expectedReady)
		return false
	}
	return true
}

// Because the v1 extensions API is not available on all supported k8s versions
// yet and because Go doesn't support generics, we need to have a duplicate
// function to support the v1beta1 types
func (c *readyChecker) crdBetaReady(crd apiextv1beta1.CustomResourceDefinition) bool {
	for _, condition := range crd.Status.Conditions {
		switch condition.Type {
		case apiextv1beta1.NamesAccepted:
			if condition.Status == apiextv1beta1.ConditionFalse {
				// This means there is a naming conflict, but this function does not need to handle it as a failure.
				// Instead, we treat it as successful so the process can continue.
				return true
			}
		case apiextv1beta1.Established:
			if condition.Status == apiextv1beta1.ConditionTrue {
				return true
			}
		// For other condition types, do nothing
		default:
		}

	}
	return false
}

func (c *readyChecker) crdReady(crd apiextv1.CustomResourceDefinition) bool {
	for _, condition := range crd.Status.Conditions {
		switch condition.Type {
		case apiextv1.NamesAccepted:
			if condition.Status == apiextv1.ConditionFalse {
				return true
			}
		case apiextv1.Established:
			if condition.Status == apiextv1.ConditionTrue {
				return true
			}
		// For other condition types, do nothing
		default:
		}
	}
	return false
}

func (c *readyChecker) statefulSetReady(stSet *appsv1.StatefulSet) bool {
	// If the update strategy is not a rolling update, there will be nothing to wait for
	if stSet.Spec.UpdateStrategy.Type != appsv1.RollingUpdateStatefulSetStrategyType {
		c.log.Infof("StatefulSet skipped ready check: %s/%s. updateStrategy is %v",
			stSet.Namespace, stSet.Name, stSet.Spec.UpdateStrategy.Type)
		return true
	}
	// Make sure the status is up-to-date with the StatefulSet changes
	if stSet.Status.ObservedGeneration < stSet.Generation {
		c.log.Infof("StatefulSet is not ready: %s/%s. update has not yet been observed", stSet.Namespace, stSet.Name)
		return false
	}
	var part int
	var rep = 1
	if stSet.Spec.UpdateStrategy.RollingUpdate != nil && stSet.Spec.UpdateStrategy.RollingUpdate.Partition != nil {
		part = int(*stSet.Spec.UpdateStrategy.RollingUpdate.Partition)
	}
	// If Replicas is set, use its value; otherwise, keep the default value
	if stSet.Spec.Replicas != nil {
		rep = int(*stSet.Spec.Replicas)
	}
	expectedReplicas := rep - part
	if int(stSet.Status.UpdatedReplicas) < expectedReplicas {
		c.log.Infof("StatefulSet is not ready: %s/%s. %d out of %d expected pods have been scheduled",
			stSet.Namespace, stSet.Name, stSet.Status.UpdatedReplicas, expectedReplicas)
		return false
	}
	if int(stSet.Status.ReadyReplicas) != rep {
		c.log.Infof("StatefulSet is not ready: %s/%s. %d out of %d expected pods are ready",
			stSet.Namespace, stSet.Name, stSet.Status.ReadyReplicas, rep)
		return false
	}
	if stSet.Status.CurrentRevision != stSet.Status.UpdateRevision {
		c.log.Infof("StatefulSet is not ready: %s/%s. currentRevision %s does not yet match updateRevision %s",
			stSet.Namespace, stSet.Name, stSet.Status.CurrentRevision, stSet.Status.UpdateRevision)
		return false
	}
	c.log.Infof("StatefulSet is ready: %s/%s. %d out of %d expected pods are ready",
		stSet.Namespace, stSet.Name, stSet.Status.ReadyReplicas, rep)
	return true
}

func getPods(ctx context.Context, client kubernetes.Interface, namespace, selector string) ([]corev1.Pod, error) {
	lst, err := client.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{
		LabelSelector: selector,
	})
	return lst.Items, err
}

// SelectorsForObject Modified version of https://github.com/kubernetes/kubernetes/blob/v1.14.1/pkg/kubectl/polymorphichelpers/helpers.go#L84
func SelectorsForObject(object runtime.Object) (selector labels.Selector, err error) {
	switch t := object.(type) {
	case *extensionsv1beta1.ReplicaSet:
		selector, err = metav1.LabelSelectorAsSelector(t.Spec.Selector)
	case *appsv1.ReplicaSet:
		selector, err = metav1.LabelSelectorAsSelector(t.Spec.Selector)
	case *appsv1.StatefulSet:
		selector, err = metav1.LabelSelectorAsSelector(t.Spec.Selector)
	case *appsv1.DaemonSet:
		selector, err = metav1.LabelSelectorAsSelector(t.Spec.Selector)
	case *appsv1.Deployment:
		selector, err = metav1.LabelSelectorAsSelector(t.Spec.Selector)
	case *appsv1beta2.ReplicaSet:
		selector, err = metav1.LabelSelectorAsSelector(t.Spec.Selector)
	case *corev1.ReplicationController:
		selector = labels.SelectorFromSet(t.Spec.Selector)
	case *batchv1.Job:
		selector, err = metav1.LabelSelectorAsSelector(t.Spec.Selector)
	case *appsv1beta1.StatefulSet:
		selector, err = metav1.LabelSelectorAsSelector(t.Spec.Selector)
	case *appsv1beta1.Deployment:
		selector, err = metav1.LabelSelectorAsSelector(t.Spec.Selector)
	case *appsv1beta2.StatefulSet:
		selector, err = metav1.LabelSelectorAsSelector(t.Spec.Selector)
	case *appsv1beta2.DaemonSet:
		selector, err = metav1.LabelSelectorAsSelector(t.Spec.Selector)
	case *extensionsv1beta1.DaemonSet:
		selector, err = metav1.LabelSelectorAsSelector(t.Spec.Selector)
	case *extensionsv1beta1.Deployment:
		selector, err = metav1.LabelSelectorAsSelector(t.Spec.Selector)
	case *appsv1beta2.Deployment:
		selector, err = metav1.LabelSelectorAsSelector(t.Spec.Selector)
	case *corev1.Service:
		// Check if the Service has a valid selector and create a label selector from it.
		if t.Spec.Selector == nil || len(t.Spec.Selector) == 0 {
			return nil, fmt.Errorf("invalid service '%s': Service is defined without a selector", t.Name)
		}
		selector = labels.SelectorFromSet(t.Spec.Selector)

	default:
		return nil, fmt.Errorf("selector for %T not implemented", object)
	}

	return selector, errors.Wrap(err, "invalid label selector")
}
